package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dto.RuleToDistCPDTO;
import com.hexinfo.dmpro.sparing.dto.TableDistCPDTO;
import com.hexinfo.dmpro.sparing.model.HdfsHeatBackup;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
@Slf4j
public class BackupTimeServiceImpl implements BackupTimeService {

    @Autowired
    private DataFilterService dataFilterService;
    @Autowired
    private DistCPService distCPService;
    @Autowired
    private LsFilterService lsFilterService;
    @Autowired
    private DistCPTimeService distCPTimeService;
    @Autowired
    private DataRefreshService dataRefreshService;

    @Override
    public void execShell(RuleToDistCPDTO ruleToDistCPDTO) {
        String sourceDataName = ruleToDistCPDTO.getSourceDataName();  //同步的源名
        String backupSort = ruleToDistCPDTO.getBackupSort();    //1：有源分区字段，2：无源分区字段（时间窗口）
        String logPath = dataFilterService.path(sourceDataName);      //获取总日志路径
        //插入日志文件缓存操作
        FileAppender heatlog = new FileAppender(new File(logPath), 100, true);
        heatlog.append(logTime("【"+sourceDataName+"】热备数据开始"));
        //1、把List<hdfs路径>整理成表级
        Map<String, List<HdfsHeatBackup>> entrys = dataFilterService.dbTableNameSort(ruleToDistCPDTO.getHdfsHeatBackups());
        heatlog.append(logTime("热备表数量："+entrys.size()+"，hdfs文件总数量："+ruleToDistCPDTO.getHdfsHeatBackups().size()));

        //------------------------------① LS过滤+数据备份同步-----------------------------------
        //LS操作结束标志给初始值
        lsFilterService.updateLsStatusMap(sourceDataName,ScanCommonConstants.ClusterName.RUNNING.value);
        //方案1：异步方法，先LS结束后热备（方案2：先LS结束后热备 方案3：按表LS+热备+比对）
        asyncFirstLs(entrys,ruleToDistCPDTO,heatlog);
        //------------------------------② 判断是否不需要热备--------------------------------
        //循环判断LS操作过滤后，需不需要同步
        while (lsFilterService.getLsStatus(sourceDataName).equals(ScanCommonConstants.ClusterName.RUNNING.value)){
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //ls过滤后，表数量不为0，进行热备
        if (lsFilterService.getLsStatus(sourceDataName).equals(ScanCommonConstants.ClusterName.CONTINUE.value)){
            //------------------------------③ 判断备份是否全部结束--------------------------------
            //热备结束状态，给一个初始值
            distCPService.updateEndFlag(sourceDataName,false);
            //循环判断这个源的热备是否全部结束
            while (!distCPService.getEndFlag(sourceDataName)){
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //删除源key
            distCPService.deleteEndFlag(sourceDataName);

            //------------------------------④ 数据同步------------------------------------
            //热备结束状态，给一个初始值
            distCPTimeService.updateEndFlag(sourceDataName,false);
            lsFilterService.dataSync(sourceDataName,heatlog);
            //------------------------------⑤ 判断同步是否全部结束--------------------------------
            //循环判断这个源的热备是否全部结束
            while (!distCPTimeService.getEndFlag(sourceDataName)){
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //删除源key
            distCPTimeService.deleteEndFlag(sourceDataName);

            //------------------------------⑥ 刷新hive+刷新impala------------------------------------
            //热备结束状态，给一个初始值
            dataRefreshService.updateEndFlag(sourceDataName,false);
            lsFilterService.dataRefresh(sourceDataName,backupSort,heatlog);
            //------------------------------⑦ 判断刷新hive+刷新impala是否全部结束--------------------------------
            //循环判断这个源的热备是否全部结束
            while (!dataRefreshService.getEndFlag(sourceDataName)){
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //删除源key
            dataRefreshService.deleteEndFlag(sourceDataName);

            //------------------------------⑧ 数据比对------------------------------------
            heatlog.append(logTime("【数据比对开始】"));
            heatlog.flush();
            lsFilterService.dataContrast(sourceDataName,heatlog,ruleToDistCPDTO.getWhereScriptDTOS());
        }
        lsFilterService.delLsStatus(sourceDataName);    //删除是否同步的key

        heatlog.append(logTime("【全部热备hdfs已完成】"));
        heatlog.flush();
    }

    /**
     * 异步方法，先LS结束后热备
     */
    private void asyncFirstLs(Map<String, List<HdfsHeatBackup>> entrys,
                         RuleToDistCPDTO ruleToDistCPDTO,FileAppender heatlog) {
        List<TableDistCPDTO> tableDistCPDTOs = new ArrayList<>();
        for (Map.Entry<String, List<HdfsHeatBackup>> entry : entrys.entrySet()) {
            //封装获得每个表的hdfs实体类
            TableDistCPDTO tableDistCPDTO = dataFilterService.getTableDistCPDTO(ruleToDistCPDTO, entry.getValue());
            //表日志文件缓存
            FileAppender tableLog = new FileAppender(new File(tableDistCPDTO.getLogPath()), 100, true);
            //调用异步方法
            lsFilterService.allLsFilter(tableDistCPDTOs,tableDistCPDTO,heatlog,tableLog,entrys.size());
        }
    }

    /**
     * 日志加日期时间
     * @param str
     * @return
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
